from fastapi import APIRouter, HTTPException, Depends
from typing import List, Dict, Any
import pandas as pd
import numpy as np
from sqlalchemy.orm import Session
from ..database import get_db
from ..models import Workflow, Node
from ..schemas import NodeExecutionRequest, NodeExecutionResponse, ExecuteNodesRequest
from .upload import convert_nan_to_none
import logging
import os
import json
from .ml_models import logistic_regression_core, random_forest_core, svm_core
from .feature_selection import feature_selection_core

router = APIRouter()
logger = logging.getLogger(__name__)

# 节点执行函数映射
NODE_EXECUTORS = {
    'csv_reader': lambda params: read_csv_file(params),
    'excel_reader': lambda params: read_excel_file(params),
    'data_cleaning': lambda data, params: clean_data(data, params),
    'data_standardization': lambda data, params: standardize_data(data, params),
    'outlier_detection': lambda data, params: detect_outliers(data, params),
    'one_hot_encoding': lambda data, params: one_hot_encode(data, params),
    'feature_selection_filter': lambda data, params: filter_features(data, params),
    'model_evaluation': lambda data, params: evaluate_model(data, params),
}

def read_csv_file(params: dict) -> dict:
    """读取CSV文件"""
    try:
        logger.info(f"CSV Reader 参数: {json.dumps(params, default=str)}")
        file_path = params.get('filePath')
        logger.info(f"尝试读取文件: {file_path}")
        
        if not file_path:
            logger.error("filePath 参数为空")
            return {
                'status': 'error',
                'message': '文件路径不能为空',
                'data': None
            }
            
        if not os.path.exists(file_path):
            logger.error(f"文件不存在: {file_path}")
            return {
                'status': 'error',
                'message': f'文件不存在: {file_path}',
                'data': None
            }
            
        logger.info(f"开始读取 CSV 文件: {file_path}")
        df = pd.read_csv(
            file_path,
            sep=params.get('separator', ','),
            encoding=params.get('encoding', 'utf-8'),
            nrows=params.get('max_rows')
        )
        logger.info(f"CSV 文件读取成功，共 {len(df)} 行, {len(df.columns)} 列")
        
        return {
            'status': 'success',
            'message': f'成功读取CSV文件，共{len(df)}行{len(df.columns)}列',
            'data': df.to_dict('records')
        }
    except Exception as e:
        logger.error(f"读取CSV文件失败: {str(e)}")
        return {
            'status': 'error',
            'message': f'读取CSV文件失败: {str(e)}',
            'data': None
        }

def read_excel_file(params: dict) -> dict:
    """读取Excel文件"""
    try:
        logger.info(f"Excel Reader 参数: {json.dumps(params, default=str)}")
        file_path = params.get('filePath')
        sheet_name = params.get('sheet_name')  # 获取工作表名称
        logger.info(f"尝试读取文件: {file_path}")
        if sheet_name:
            logger.info(f"指定工作表: {sheet_name}")
        
        if not file_path:
            logger.error("filePath 参数为空")
            return {
                'status': 'error',
                'message': '文件路径不能为空',
                'data': None
            }
            
        if not os.path.exists(file_path):
            logger.error(f"文件不存在: {file_path}")
            return {
                'status': 'error',
                'message': f'文件不存在: {file_path}',
                'data': None
            }
            
        logger.info(f"开始读取 Excel 文件: {file_path}")
        
        # 获取工作表名称
        all_sheets = []
        try:
            excel_file = pd.ExcelFile(file_path)
            all_sheets = excel_file.sheet_names
            current_sheet = sheet_name or all_sheets[0]
            logger.info(f"文件包含工作表: {', '.join(all_sheets)}, 当前使用: {current_sheet}")
        except Exception as e:
            logger.warning(f"获取工作表列表失败: {str(e)}")
        
        # 设置Excel读取参数
        excel_params = {}
        if sheet_name:
            logger.info(f"使用指定工作表: {sheet_name}")
            excel_params['sheet_name'] = sheet_name
        
        # 读取Excel文件
        df = pd.read_excel(file_path, **excel_params)
        logger.info(f"成功读取 Excel 文件: {file_path}, 工作表: {sheet_name or '默认'}, 行数={len(df)}")
        
        node_result = {
            "status": "success", 
            "message": f"成功读取 Excel 文件 '{current_sheet}' 工作表，共 {len(df)} 行", 
            "data": convert_nan_to_none(df.to_dict('records')),
            "columns": df.columns.tolist(),
            "row_count": len(df),
            "sheet_name": current_sheet,
            "available_sheets": all_sheets
        }
        # 存储完整数据供后续节点使用
        node_outputs[node_id] = {
            "data": df,
            "columns": df.columns.tolist(),
            "feature_importance": df.columns.tolist(),
            "sheet_name": current_sheet,
            "available_sheets": all_sheets
        }
        
        return node_result
    except Exception as e:
        logger.error(f"读取Excel文件失败: {str(e)}")
        import traceback
        logger.error(f"详细错误信息: {traceback.format_exc()}")
        return {
            'status': 'error',
            'message': f'读取Excel文件失败: {str(e)}',
            'data': None
        }

def clean_data(data: pd.DataFrame, params: dict) -> pd.DataFrame:
    """数据清洗函数"""
    cleaning_type = params.get('cleaning_type', 'missing')
    columns = params.get('columns', '').split(',')
    
    if cleaning_type == 'missing':
        strategy = params.get('strategy', 'mean')
        if strategy == 'mean':
            return data[columns].fillna(data[columns].mean())
        elif strategy == 'median':
            return data[columns].fillna(data[columns].median())
        elif strategy == 'mode':
            return data[columns].fillna(data[columns].mode().iloc[0])
        elif strategy == 'constant':
            fill_value = params.get('fill_value', '')
            return data[columns].fillna(fill_value)
    
    return data

def standardize_data(data: pd.DataFrame, params: dict) -> pd.DataFrame:
    """数据标准化函数"""
    method = params.get('method', 'zscore')
    columns_raw = params.get('columns', [])
    
    # 处理列选择
    if isinstance(columns_raw, list):
        columns = columns_raw
    elif isinstance(columns_raw, str):
        columns = [col.strip() for col in columns_raw.split(',') if col.strip()]
    else:
        columns = []
    
    # 如果没有指定列，默认处理所有数值列
    if not columns:
        columns = [col for col in data.columns if pd.api.types.is_numeric_dtype(data[col])]
        logger.info(f"未指定列，使用所有数值列: {columns}")
    
    # 检查列是否存在于数据中
    valid_columns = [col for col in columns if col in data.columns]
    if not valid_columns:
        logger.warning(f"标准化函数未找到有效列: {columns}")
        return data
    
    # 复制原始数据
    result = data.copy()
    
    try:
        logger.info(f"开始标准化数据，方法: {method}, 列: {valid_columns}")
        
        if method == 'zscore':
            # Z-Score标准化（标准化）
            with_mean = params.get('with_mean', 'true') == 'true'
            with_std = params.get('with_std', 'true') == 'true'
            
            logger.info(f"Z-Score标准化参数: with_mean={with_mean}, with_std={with_std}")
            
            for col in valid_columns:
                series = data[col]
                # 只处理数值列
                if pd.api.types.is_numeric_dtype(series):
                    # 中心化（减去均值）
                    if with_mean:
                        result[col] = series - series.mean()
                    else:
                        # 如果不中心化，确保复制原始数据
                        result[col] = series.copy()
                    # 缩放（除以标准差）
                    if with_std:
                        std = series.std()
                        if std > 0:  # 避免除以零
                            result[col] = result[col] / std
                else:
                    logger.warning(f"跳过非数值列: {col}")
                    
        elif method == 'minmax':
            # Min-Max标准化（归一化）
            feature_range_str = params.get('feature_range', '0,1')
            try:
                min_val, max_val = map(float, feature_range_str.split(','))
                logger.info(f"Min-Max标准化参数: feature_range=[{min_val}, {max_val}]")
                
                for col in valid_columns:
                    series = data[col]
                    # 只处理数值列
                    if pd.api.types.is_numeric_dtype(series):
                        col_min = series.min()
                        col_max = series.max()
                        if col_max > col_min:  # 避免除以零
                            result[col] = (series - col_min) / (col_max - col_min) * (max_val - min_val) + min_val
                    else:
                        logger.warning(f"跳过非数值列: {col}")
            except ValueError as e:
                logger.error(f"解析feature_range参数错误: {e}")
                
        elif method == 'robust':
            # Robust标准化（使用四分位数代替均值和标准差）
            quantile_range_str = params.get('quantile_range', '25.0,75.0')
            try:
                q_low, q_high = map(float, quantile_range_str.split(','))
                q_low, q_high = q_low / 100.0, q_high / 100.0  # 转换为百分比
                logger.info(f"Robust标准化参数: quantile_range=[{q_low}, {q_high}]")
                
                for col in valid_columns:
                    series = data[col]
                    # 只处理数值列
                    if pd.api.types.is_numeric_dtype(series):
                        q_min = series.quantile(q_low)
                        q_max = series.quantile(q_high)
                        if q_max > q_min:  # 避免除以零
                            result[col] = (series - series.median()) / (q_max - q_min)
                    else:
                        logger.warning(f"跳过非数值列: {col}")
            except ValueError as e:
                logger.error(f"解析quantile_range参数错误: {e}")
        
        logger.info(f"数据标准化完成，返回 {len(result)} 行数据")
        return result
    except Exception as e:
        logger.exception(f"标准化数据时出错: {str(e)}")
        return data

def detect_outliers(data: pd.DataFrame, params: dict) -> pd.DataFrame:
    """异常值检测函数"""
    method = params.get('outlier_method', 'z-score')
    columns = params.get('columns', '').split(',')
    threshold = params.get('z_score_threshold', 3)
    
    if method == 'z-score':
        z_scores = np.abs((data[columns] - data[columns].mean()) / data[columns].std())
        return data[z_scores < threshold]
    elif method == 'iqr':
        iqr_multiplier = params.get('iqr_multiplier', 1.5)
        Q1 = data[columns].quantile(0.25)
        Q3 = data[columns].quantile(0.75)
        IQR = Q3 - Q1
        return data[~((data[columns] < (Q1 - iqr_multiplier * IQR)) | (data[columns] > (Q3 + iqr_multiplier * IQR))).any(axis=1)]
    
    return data

def convert_node_ids_to_str(node_ids: List[str]) -> List[str]:
    """将节点ID转换为字符串"""
    return [str(node_id) for node_id in node_ids]

@router.post("/execute-nodes")
async def execute_nodes(
    execute_request: ExecuteNodesRequest,
    db: Session = Depends(get_db)
):
    """执行工作流中指定的节点"""
    try:
        # 记录完整请求内容
        #logger.info(f"收到执行节点请求: {json.dumps(execute_request.dict(), default=str, ensure_ascii=False)}")
        
        workflow_id = execute_request.workflow_id
        node_ids = execute_request.node_ids
        nodes_data = execute_request.nodes or []
        
        logger.info(f"执行请求: workflow_id={workflow_id}, node_ids={node_ids}")
        if nodes_data:
            #logger.info(f"提供的节点数据: {json.dumps(nodes_data, default=str)}")
            pass
    except Exception as e:
        logger.error(f"请求解析失败: {str(e)}")
        logger.exception("请求解析错误详情:")
        raise HTTPException(status_code=422, detail=f"请求格式错误: {str(e)}")
    
    # 获取工作流
    workflow = db.query(Workflow).filter(Workflow.id == workflow_id).first()
    if not workflow:
        logger.error(f"未找到工作流 ID={workflow_id}")
        raise HTTPException(status_code=404, detail=f"未找到工作流 ID={workflow_id}")
    
    logger.info(f"找到工作流: {workflow.name}")
    
    # 将节点 ID 转换为字符串以便匹配
    str_node_ids = [str(nid) for nid in node_ids]
    logger.info(f"要执行的节点 IDs (字符串格式): {str_node_ids}")
    
    # 使用工作流的节点和边数据
    workflow_nodes = workflow.nodes or []
    workflow_edges = workflow.edges or []
    logger.debug(f"工作流节点: {workflow_nodes}")
    logger.debug(f"工作流边: {workflow_edges}")
    
    # 构建节点依赖关系图
    edges_dict = {}
    logger.info(f"构建节点依赖关系图，边的数量: {len(workflow_edges)}")
    for edge in workflow_edges:
        source = str(edge.get('source'))
        target = str(edge.get('target'))
        
        logger.info(f"边连接: source={source} -> target={target}")
        
        if target not in edges_dict:
            edges_dict[target] = []
        
        edges_dict[target].append(source)

    logger.info(f"节点依赖关系图: {edges_dict}")
    
    # 查找匹配的节点
    nodes_to_execute = []
    node_data_map = {}
    
    if nodes_data:
        # 使用前端提供的节点数据
        for node_data in nodes_data:
            node_id = str(node_data.get('id'))
            if node_id in str_node_ids:
                logger.info(f"使用前端提供的节点数据: ID={node_id}")
                nodes_to_execute.append(node_data)
                node_data_map[node_id] = node_data
    
    # 如果前端没有提供完整的节点数据，则从工作流数据中获取
    if len(nodes_to_execute) < len(str_node_ids):
        for node in workflow_nodes:
            node_id = str(node.get('id'))
            if node_id in str_node_ids and node_id not in node_data_map:
                logger.info(f"从工作流中获取节点数据: ID={node_id}")
                nodes_to_execute.append(node)
                node_data_map[node_id] = node
    
    if not nodes_to_execute:
        logger.error(f"未找到要执行的节点 IDs={str_node_ids}")
        raise HTTPException(status_code=404, detail=f"未找到要执行的节点 IDs={str_node_ids}")
    
    logger.info(f"找到 {len(nodes_to_execute)} 个节点要执行")
    
    # 使用拓扑排序对节点进行排序，确保上游节点先执行
    execution_order = topological_sort(nodes_to_execute, edges_dict)
    
    if execution_order:
        ordered_nodes = [node_data_map[node_id] for node_id in execution_order if node_id in node_data_map]
        logger.info(f"节点执行顺序: {execution_order}")
        nodes_to_execute = ordered_nodes
    else:
        logger.warning("无法确定节点的拓扑排序，可能存在循环依赖，将使用原始顺序执行")
    
    # 执行节点
    results = {}
    node_outputs = {}  # 存储每个节点的输出数据，用于后续节点的输入

    # 预填充 node_outputs，优先使用前端传递的 lastExecutionResult 数据
    for node in nodes_to_execute:
        node_id = str(node.get('id'))
        node_data = node.get('data', {})
        last_result = node_data.get('lastExecutionResult')
        logger.info(f"检查节点 {node_id} 类型 {node_data.get('type')} 的lastExecutionResult: {last_result is not None}")
        
        # 特别处理机器学习节点的lastExecutionResult
        if node_data.get('type') in ['logistic_regression', 'random_forest', 'svm'] and last_result:
            logger.info(f"机器学习节点 {node_id} 的lastExecutionResult: {list(last_result.keys()) if isinstance(last_result, dict) else 'not a dict'}")
            if isinstance(last_result, dict) and 'metrics' in last_result:
                logger.info(f"保存机器学习节点 {node_id} 的metrics数据供后续节点使用")
                if node_id not in node_outputs:
                    node_outputs[node_id] = {
                        'metrics': last_result.get('metrics'),
                        'training_info': last_result.get('training_info'),
                        'feature_importance': last_result.get('feature_importance')
                    }
                    # 如果有数据，也添加进去
                    if 'data' in last_result:
                        node_outputs[node_id]['data'] = pd.DataFrame(last_result['data'])
                        node_outputs[node_id]['columns'] = last_result.get('columns', [])
        
        # 常规数据节点处理
        elif last_result and isinstance(last_result, dict) and 'data' in last_result:
            # 只填充一次，避免覆盖后续执行结果
            if node_id not in node_outputs:
                node_outputs[node_id] = {
                    'data': pd.DataFrame(last_result['data']) if isinstance(last_result['data'], list) else last_result['data'],
                    'columns': last_result.get('columns', []),
                    'feature_importance': last_result.get('feature_importance', None)
                }
    
    # 特别处理模型评估节点，确保它能访问到上游节点的数据
    for node in nodes_to_execute:
        node_id = str(node.get('id'))
        node_data = node.get('data', {})
        if node_data.get('type') == 'model_evaluation':
            logger.info(f"找到模型评估节点 {node_id}，准备确保它能访问上游节点数据")
            if node_id in edges_dict:
                for source_id in edges_dict[node_id]:
                    source_node = node_data_map.get(source_id)
                    if source_node:
                        source_data = source_node.get('data', {})
                        source_type = source_data.get('type')
                        logger.info(f"上游节点 {source_id} 类型为 {source_type}")
                        
                        # 如果上游是机器学习节点但还没有输出，提前尝试获取其数据
                        if source_type in ['logistic_regression', 'random_forest', 'svm'] and source_id not in node_outputs:
                            logger.info(f"上游机器学习节点 {source_id} 尚未有输出，尝试提前获取")
                            last_result = source_data.get('lastExecutionResult', {})
                            if last_result and 'metrics' in last_result:
                                logger.info(f"成功获取上游节点 {source_id} 的metrics数据")
                                node_outputs[source_id] = {
                                    'metrics': last_result.get('metrics'),
                                    'training_info': last_result.get('training_info'),
                                    'feature_importance': last_result.get('feature_importance')
                                }
                                logger.info(f"已为模型评估节点 {node_id} 准备好上游节点 {source_id} 的数据")
                        
                        # 检查最后执行结果是否被正确保存到节点输出
                        if source_id in node_outputs:
                            logger.info(f"上游节点 {source_id} 已有输出: {list(node_outputs[source_id].keys())}")
                        else:
                            logger.warning(f"上游节点 {source_id} 尚无输出数据")

    for node in nodes_to_execute:
        node_id = str(node.get('id'))
        logger.info(f"准备执行节点 ID={node_id}")
        
        try:
            # 获取节点类型和参数
            node_type = node.get('data', {}).get('type')
            node_params = node.get('data', {}).get('params', {})
            
            #logger.info(f"节点 ID={node_id}, 类型={node_type}, 参数={node_params}")
            
            # 获取上游节点的数据
            upstream_data = None
            if node_id in edges_dict:
                logger.info(f"节点 {node_id} 有上游依赖节点: {edges_dict[node_id]}")
                for source_id in edges_dict[node_id]:
                    # 检查并输出所有可能的节点输出，帮助调试
                    logger.info(f"检查上游节点 {source_id} 是否在node_outputs中: {source_id in node_outputs}")
                    if source_id in node_outputs:
                        upstream_data = node_outputs[source_id]
                        logger.info(f"使用上游节点 ID={source_id} 的数据作为输入")
                        logger.info(f"上游数据Keys: {list(upstream_data.keys()) if upstream_data else None}")
                        logger.info(f"上游数据的DataFrame形状: {upstream_data.get('data').shape if upstream_data and 'data' in upstream_data and hasattr(upstream_data.get('data'), 'shape') else 'not a DataFrame'}")
                        break
                    else:
                        logger.warning(f"上游节点 ID={source_id} 没有输出数据")
                        # 检查所有节点的data和lastExecutionResult，帮助诊断
                        for n in nodes_to_execute:
                            if str(n.get('id')) == source_id:
                                logger.info(f"找到上游节点 {source_id} 的数据:")
                                logger.info(f"节点类型: {n.get('data', {}).get('type')}")
                                logger.info(f"是否有lastExecutionResult: {bool(n.get('data', {}).get('lastExecutionResult'))}")
                                last_exec = n.get('data', {}).get('lastExecutionResult', {})
                                logger.info(f"lastExecutionResult keys: {list(last_exec.keys()) if last_exec else None}")
                                logger.info(f"lastExecutionResult是否有data: {'data' in last_exec if last_exec else False}")
                                if last_exec and 'data' in last_exec:
                                    logger.info(f"lastExecutionResult.data类型: {type(last_exec['data'])}")
                                    logger.info(f"lastExecutionResult.data长度: {len(last_exec['data']) if isinstance(last_exec['data'], list) else 'not a list'}")
                                
                                # 尝试直接从上游节点的lastExecutionResult创建upstream_data
                                if last_exec and 'data' in last_exec and isinstance(last_exec['data'], list) and len(last_exec['data']) > 0:
                                    logger.info(f"尝试直接使用上游节点 {source_id} 的lastExecutionResult.data")
                                    try:
                                        # 创建DataFrame并保存到node_outputs
                                        node_outputs[source_id] = {
                                            'data': pd.DataFrame(last_exec['data']),
                                            'columns': last_exec.get('columns', list(last_exec['data'][0].keys())),
                                            'feature_importance': last_exec.get('feature_importance', None)
                                        }
                                        upstream_data = node_outputs[source_id]
                                        logger.info(f"成功直接使用lastExecutionResult创建上游数据，形状: {upstream_data['data'].shape}")
                                        break
                                    except Exception as e:
                                        logger.exception(f"尝试使用lastExecutionResult时出错: {str(e)}")
            else:
                logger.warning(f"节点 {node_id} 没有上游依赖节点，edges_dict={edges_dict}")
            
            # 执行节点
            if node_type in ['csv_reader', 'excel_reader']:
                # 检查文件路径
                file_path = node_params.get('filePath')
                if not file_path:
                    logger.error(f"节点 ID={node_id} 缺少文件路径")
                    results[node_id] = {
                        "status": "error", 
                        "message": "缺少文件路径"
                    }
                    continue
                
                # 检查文件是否存在
                if not os.path.exists(file_path):
                    logger.error(f"节点 ID={node_id} 的文件不存在: {file_path}")
                    results[node_id] = {
                        "status": "error", 
                        "message": f"文件不存在: {file_path}"
                    }
                    continue
                
                logger.info(f"读取文件: {file_path}")
                
                try:
                    # 执行数据读取
                    if node_type == 'csv_reader':
                        df = pd.read_csv(file_path)
                        logger.info(f"成功读取 CSV 文件: {file_path}, 行数={len(df)}")
                        node_result = {
                            "status": "success", 
                            "message": f"成功读取 CSV 文件，共 {len(df)} 行", 
                            # 修改此处，去掉 head(5) 限制
                            "data": convert_nan_to_none(df.to_dict('records')),
                            "columns": df.columns.tolist(),
                            "row_count": len(df)
                        }
                        # 存储完整数据供后续节点使用
                        node_outputs[node_id] = {
                            "data": df,
                            "columns": df.columns.tolist(),
                            "feature_importance": df.columns.tolist()
                        }
                    elif node_type == 'excel_reader':
                        # 获取工作表名称
                        sheet_name = node_params.get('sheet_name')
                        
                        # 设置Excel读取参数
                        excel_params = {}
                        if sheet_name:
                            logger.info(f"使用指定工作表: {sheet_name}")
                            excel_params['sheet_name'] = sheet_name
                        
                        # 读取Excel文件
                        df = pd.read_excel(file_path, **excel_params)
                        logger.info(f"成功读取 Excel 文件: {file_path}, 工作表: {sheet_name or '默认'}, 行数={len(df)}")
                        
                        # 获取所有工作表名称
                        all_sheets = []
                        try:
                            excel_file = pd.ExcelFile(file_path)
                            all_sheets = excel_file.sheet_names
                            current_sheet = sheet_name or all_sheets[0]
                            logger.info(f"文件包含工作表: {', '.join(all_sheets)}, 当前使用: {current_sheet}")
                        except Exception as e:
                            logger.warning(f"获取工作表列表失败: {str(e)}")
                        
                        node_result = {
                            "status": "success", 
                            "message": f"成功读取 Excel 文件 '{current_sheet}' 工作表，共 {len(df)} 行", 
                            "data": convert_nan_to_none(df.to_dict('records')),
                            "columns": df.columns.tolist(),
                            "row_count": len(df),
                            "sheet_name": current_sheet,
                            "available_sheets": all_sheets
                        }
                        # 存储完整数据供后续节点使用
                        node_outputs[node_id] = {
                            "data": df,
                            "columns": df.columns.tolist(),
                            "feature_importance": df.columns.tolist(),
                            "sheet_name": current_sheet,
                            "available_sheets": all_sheets
                        }
                    results[node_id] = node_result
                except Exception as e:
                    logger.exception(f"读取文件失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"读取文件失败: {str(e)}"
                    }
            elif node_type == 'structured_view':
                # 结构化视图节点依赖上游数据
                if not upstream_data:
                    logger.error(f"结构化视图节点 ID={node_id} 没有上游数据")
                    results[node_id] = {
                        "status": "error",
                        "message": "没有可显示的数据，请确保连接到了数据源节点"
                    }
                    continue
                
                try:
                    # 获取上游数据
                    df = upstream_data.get("data")
                    columns = upstream_data.get("columns", [])
                    
                    if df is None or len(df) == 0:
                        logger.warning(f"结构化视图节点 ID={node_id} 的上游数据为空")
                        results[node_id] = {
                            "status": "warning",
                            "message": "数据源为空",
                            "data": [],
                            "columns": columns
                        }
                    else:
                        # 去掉数据量限制
                        logger.info(f"结构化视图: 返回全部 {len(df)} 行数据")
                        
                        results[node_id] = {
                            "status": "success",
                            "message": f"成功获取数据，共 {len(df)} 行",
                            "data": convert_nan_to_none(df.to_dict('records')),
                            "columns": columns,
                            "feature_importance": columns
                        }
                        # 将数据传递给后续节点
                        node_outputs[node_id] = upstream_data
                except Exception as e:
                    logger.exception(f"处理结构化视图数据失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"处理数据失败: {str(e)}"
                    }
            elif node_type == 'logistic_regression':
                # 逻辑回归节点依赖上游数据
                if not upstream_data:
                    logger.error(f"逻辑回归节点 ID={node_id} 没有上游数据")
                    results[node_id] = {
                        "status": "error",
                        "message": "没有可用于训练的数据，请确保连接到了数据源节点"
                    }
                    continue
                try:
                    logger.info(f"逻辑回归节点 ID={node_id} 的上游数据: {upstream_data.keys()}")
                    if isinstance(upstream_data.get('data'), pd.DataFrame):
                        df = upstream_data.get('data')
                        logger.info(f"上游数据是DataFrame，形状: {df.shape}")
                    elif isinstance(upstream_data.get('data'), list) and len(upstream_data.get('data', [])) > 0:
                        df = pd.DataFrame(upstream_data.get('data'))
                        logger.info(f"已将字典列表转换为DataFrame，形状: {df.shape}")
                    else:
                        logger.error(f"上游数据格式不正确: {type(upstream_data.get('data'))}")
                        results[node_id] = {
                            "status": "error",
                            "message": "上游数据格式不正确"
                        }
                        continue
                    target_column = node_params.get('target_column')
                    feature_columns = node_params.get('feature_columns', [])
                    if isinstance(feature_columns, str):
                        feature_columns = feature_columns.split(',')
                        logger.info(f"将feature_columns从字符串转换为列表: {feature_columns}")
                    logger.info(f"目标列: {target_column}, 特征列: {feature_columns}")
                    logger.info(f"DataFrame列: {df.columns.tolist()}")
                    try:
                        metrics, feature_importance, training_info = logistic_regression_core(
                            df,
                            feature_columns,
                            target_column,
                            C=float(node_params.get('C', 1.0)),
                            max_iter=int(node_params.get('max_iter', 100)),
                            penalty=node_params.get('penalty', 'l2'),
                            test_size=float(node_params.get('test_size', 0.2)),
                            random_state=42
                        )
                    except Exception as e:
                        logger.exception(f"逻辑回归核心函数执行失败: {str(e)}")
                        results[node_id] = {
                            "status": "error",
                            "message": f"逻辑回归模型训练失败: {str(e)}"
                        }
                        continue
                    results[node_id] = {
                        "status": "success",
                        "message": "逻辑回归模型训练成功",
                        "metrics": metrics,
                        "feature_importance": feature_importance,
                        "training_info": training_info
                    }
                    node_outputs[node_id] = {
                        "data": df,
                        "columns": df.columns.tolist(),
                        "metrics": metrics,
                        "feature_importance": feature_importance,
                        "training_info": training_info
                    }
                    logger.info(f"逻辑回归节点 {node_id} 输出已保存，包含以下数据: {list(node_outputs[node_id].keys())}")
                except Exception as e:
                    logger.exception(f"逻辑回归节点 ID={node_id} 执行失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"逻辑回归模型训练失败: {str(e)}"
                    }
            elif node_type == 'random_forest':
                # 随机森林节点依赖上游数据
                if not upstream_data:
                    logger.error(f"随机森林节点 ID={node_id} 没有上游数据")
                    results[node_id] = {
                        "status": "error",
                        "message": "没有可用于训练的数据，请确保连接到了数据源节点"
                    }
                    continue
                try:
                    logger.info(f"随机森林节点 ID={node_id} 的上游数据: {upstream_data.keys()}")
                    if isinstance(upstream_data.get('data'), pd.DataFrame):
                        df = upstream_data.get('data')
                        logger.info(f"上游数据是DataFrame，形状: {df.shape}")
                    elif isinstance(upstream_data.get('data'), list) and len(upstream_data.get('data', [])) > 0:
                        df = pd.DataFrame(upstream_data.get('data'))
                        logger.info(f"已将字典列表转换为DataFrame，形状: {df.shape}")
                    else:
                        logger.error(f"上游数据格式不正确: {type(upstream_data.get('data'))}")
                        results[node_id] = {
                            "status": "error",
                            "message": "上游数据格式不正确"
                        }
                        continue
                    target_column = node_params.get('target_column')
                    feature_columns = node_params.get('feature_columns', [])
                    if isinstance(feature_columns, str):
                        feature_columns = feature_columns.split(',')
                        logger.info(f"将feature_columns从字符串转换为列表: {feature_columns}")
                    logger.info(f"目标列: {target_column}, 特征列: {feature_columns}")
                    logger.info(f"DataFrame列: {df.columns.tolist()}")
                    try:
                        metrics, feature_importance, training_info = random_forest_core(
                            df,
                            feature_columns,
                            target_column,
                            n_estimators=int(node_params.get('n_estimators', 100)),
                            max_depth=int(node_params.get('max_depth')) if node_params.get('max_depth') else None,
                            min_samples_split=int(node_params.get('min_samples_split', 2)),
                            min_samples_leaf=int(node_params.get('min_samples_leaf', 1)),
                            criterion=node_params.get('criterion', 'gini'),
                            test_size=float(node_params.get('test_size', 0.2)),
                            random_state=int(node_params.get('random_state', 42))
                        )
                    except Exception as e:
                        logger.exception(f"随机森林核心函数执行失败: {str(e)}")
                        results[node_id] = {
                            "status": "error",
                            "message": f"随机森林模型训练失败: {str(e)}"
                        }
                        continue
                    results[node_id] = {
                        "status": "success",
                        "message": "随机森林模型训练成功",
                        "metrics": metrics,
                        "feature_importance": feature_importance,
                        "training_info": training_info
                    }
                    node_outputs[node_id] = {
                        "data": df,
                        "columns": df.columns.tolist(),
                        "metrics": metrics,
                        "feature_importance": feature_importance,
                        "training_info": training_info
                    }
                    logger.info(f"随机森林节点 {node_id} 输出已保存，包含以下数据: {list(node_outputs[node_id].keys())}")
                except Exception as e:
                    logger.exception(f"随机森林节点 ID={node_id} 执行失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"随机森林模型训练失败: {str(e)}"
                    }
            elif node_type == 'svm':
                # SVM节点依赖上游数据
                if not upstream_data:
                    logger.error(f"SVM节点 ID={node_id} 没有上游数据")
                    results[node_id] = {
                        "status": "error",
                        "message": "没有可用于训练的数据，请确保连接到了数据源节点"
                    }
                    continue
                try:
                    logger.info(f"SVM节点 ID={node_id} 的上游数据: {upstream_data.keys()}")
                    if isinstance(upstream_data.get('data'), pd.DataFrame):
                        df = upstream_data.get('data')
                        logger.info(f"上游数据是DataFrame，形状: {df.shape}")
                    elif isinstance(upstream_data.get('data'), list) and len(upstream_data.get('data', [])) > 0:
                        df = pd.DataFrame(upstream_data.get('data'))
                        logger.info(f"已将字典列表转换为DataFrame，形状: {df.shape}")
                    else:
                        logger.error(f"上游数据格式不正确: {type(upstream_data.get('data'))}")
                        results[node_id] = {
                            "status": "error",
                            "message": "上游数据格式不正确"
                        }
                        continue
                    target_column = node_params.get('target_column')
                    feature_columns = node_params.get('feature_columns', [])
                    if isinstance(feature_columns, str):
                        feature_columns = feature_columns.split(',')
                        logger.info(f"将feature_columns从字符串转换为列表: {feature_columns}")
                    logger.info(f"目标列: {target_column}, 特征列: {feature_columns}")
                    logger.info(f"DataFrame列: {df.columns.tolist()}")
                    
                    # 处理gamma参数
                    gamma_param = node_params.get('gamma', 'scale')
                    if gamma_param not in ['scale', 'auto'] and gamma_param is not None:
                        try:
                            gamma_param = float(gamma_param)
                        except ValueError:
                            gamma_param = 'scale'
                            logger.warning(f"无效的gamma参数值，使用默认值'scale'")
                    
                    try:
                        metrics, feature_importance, training_info = svm_core(
                            df,
                            feature_columns,
                            target_column,
                            C=float(node_params.get('C', 1.0)),
                            kernel=node_params.get('kernel', 'rbf'),
                            gamma=gamma_param,
                            degree=int(node_params.get('degree', 3)) if node_params.get('kernel') == 'poly' else 3,
                            test_size=float(node_params.get('test_size', 0.2)),
                            random_state=int(node_params.get('random_state', 42))
                        )
                    except Exception as e:
                        logger.exception(f"SVM核心函数执行失败: {str(e)}")
                        results[node_id] = {
                            "status": "error",
                            "message": f"SVM模型训练失败: {str(e)}"
                        }
                        continue
                    results[node_id] = {
                        "status": "success",
                        "message": "SVM模型训练成功",
                        "metrics": metrics,
                        "feature_importance": feature_importance,
                        "training_info": training_info
                    }
                    node_outputs[node_id] = {
                        "data": df,
                        "columns": df.columns.tolist(),
                        "metrics": metrics,
                        "feature_importance": feature_importance,
                        "training_info": training_info
                    }
                    logger.info(f"SVM节点 {node_id} 输出已保存，包含以下数据: {list(node_outputs[node_id].keys())}")
                except Exception as e:
                    logger.exception(f"SVM节点 ID={node_id} 执行失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"SVM模型训练失败: {str(e)}"
                    }
            elif node_type == 'model_evaluation':
                # 模型评估节点依赖上游数据
                if not upstream_data:
                    logger.error(f"模型评估节点 ID={node_id} 没有上游数据")
                    logger.error(f"节点依赖关系: {edges_dict.get(node_id, [])}")
                    logger.error(f"可用的节点输出: {list(node_outputs.keys())}")
                    
                    # 在没有upstream_data的情况下，尝试直接从node_outputs中获取上游节点的metrics
                    upstream_found = False
                    if node_id in edges_dict:
                        for source_id in edges_dict[node_id]:
                            if source_id in node_outputs:
                                logger.info(f"尝试直接使用node_outputs中的上游节点 {source_id} 数据")
                                upstream_data = node_outputs[source_id]
                                upstream_found = True
                                logger.info(f"成功获取上游节点 {source_id} 数据: {list(upstream_data.keys())}")
                                break
                    
                    # 如果仍未找到上游数据，尝试从任何机器学习节点的输出中获取metrics
                    if not upstream_found:
                        logger.info("尝试从任何机器学习节点获取metrics数据")
                        for out_id, output in node_outputs.items():
                            if isinstance(output, dict) and 'metrics' in output:
                                logger.info(f"找到节点 {out_id} 的metrics数据")
                                upstream_data = output
                                upstream_found = True
                                break
                    
                    # 如果仍未找到数据，返回错误
                    if not upstream_found:
                        results[node_id] = {
                            "status": "error",
                            "message": "没有可用于评估的数据，请确保连接到了模型训练节点"
                        }
                        continue
                
                try:
                    # 检查上游节点数据中是否包含机器学习模型的评估指标
                    metrics = None
                    model = None
                    training_info = None
                    feature_importance = None
                    
                    # 记录上游数据内容
                    logger.info(f"模型评估节点 ID={node_id} 的上游数据: {list(upstream_data.keys()) if upstream_data else None}")
                    if 'metrics' in upstream_data:
                        logger.info("上游数据中直接包含metrics")
                        metrics = upstream_data.get('metrics')
                        training_info = upstream_data.get('training_info')
                        feature_importance = upstream_data.get('feature_importance')
                    
                    # 如果上游数据中没有metrics，尝试从其他来源获取
                    if not metrics:
                        # 首先尝试从上游节点的lastExecutionResult中获取metrics
                        for source_id in edges_dict.get(node_id, []):
                            logger.info(f"检查上游节点 {source_id} 是否有metrics数据")
                            source_node = node_data_map.get(source_id)
                            if source_node:
                                source_data = source_node.get('data', {})
                                last_result = source_data.get('lastExecutionResult', {})
                                logger.info(f"上游节点 {source_id} 的lastExecutionResult: {list(last_result.keys()) if last_result else None}")
                                
                                if 'metrics' in last_result:
                                    logger.info(f"从上游节点 {source_id} 获取到metrics数据")
                                    metrics = last_result.get('metrics')
                                    training_info = last_result.get('training_info')
                                    feature_importance = last_result.get('feature_importance')
                                    break
                    
                    # 如果没有找到metrics，再尝试从所有节点中查找
                    if not metrics:
                        logger.info("尝试查找任何包含metrics的节点")
                        for n_id, n_output in node_outputs.items():
                            if isinstance(n_output, dict) and 'metrics' in n_output:
                                logger.info(f"找到节点 {n_id} 的metrics数据")
                                metrics = n_output.get('metrics')
                                training_info = n_output.get('training_info')
                                feature_importance = n_output.get('feature_importance')
                                break
                    
                    if not metrics:
                        logger.error(f"模型评估节点 ID={node_id} 没有找到上游节点的评估指标")
                        results[node_id] = {
                            "status": "error",
                            "message": "没有找到上游模型节点的评估指标，请确保机器学习模型节点已成功执行"
                        }
                        continue
                    
                    # 记录ROC曲线数据情况
                    if 'roc_curve' in metrics:
                        logger.info(f"模型评估生成ROC曲线数据: AUC={metrics['roc_curve']['auc']}")
                    else:
                        logger.info(f"模型评估未生成ROC曲线数据")
                    
                    # 记录特征重要性数据情况
                    if feature_importance:
                        feature_count = len(feature_importance) if isinstance(feature_importance, dict) else 0
                        logger.info(f"模型评估包含特征重要性数据: {feature_count}个特征")
                    else:
                        logger.info(f"模型评估未包含特征重要性数据")
                    
                    # 准备评估结果
                    evaluation_result = {
                        "status": "success",
                        "message": "模型评估成功",
                        "metrics": metrics
                    }
                    
                    if training_info:
                        evaluation_result["training_info"] = training_info
                    if feature_importance:
                        evaluation_result["feature_importance"] = feature_importance
                    
                    results[node_id] = evaluation_result
                    
                    # 将评估结果保存到node_outputs
                    node_outputs[node_id] = {
                        "metrics": metrics,
                        "training_info": training_info,
                        "feature_importance": feature_importance
                    }
                    
                except Exception as e:
                    logger.exception(f"处理模型评估数据失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"处理数据失败: {str(e)}"
                    }
            elif node_type == 'data_standardization':
                # 数据标准化节点
                if not upstream_data:
                    logger.error(f"数据标准化节点 ID={node_id} 没有上游数据")
                    results[node_id] = {
                        "status": "error",
                        "message": "没有可用于标准化的数据，请确保连接到了数据源节点"
                    }
                    continue
                
                try:
                    logger.info(f"数据标准化节点 ID={node_id} 的上游数据: {list(upstream_data.keys())}")
                    
                    if 'data' not in upstream_data or upstream_data['data'] is None:
                        raise ValueError("上游节点没有提供数据")
                    
                    # 获取上游数据
                    if isinstance(upstream_data.get('data'), pd.DataFrame):
                        df = upstream_data.get('data')
                    elif isinstance(upstream_data.get('data'), list) and len(upstream_data.get('data', [])) > 0:
                        df = pd.DataFrame(upstream_data.get('data'))
                    else:
                        raise ValueError(f"上游数据格式不正确: {type(upstream_data.get('data'))}")
                    
                    logger.info(f"开始处理数据标准化，数据形状: {df.shape}")
                    
                    # 执行标准化
                    standardized_df = standardize_data(df, node_params)
                    
                    # 获取数据列
                    columns = standardized_df.columns.tolist()
                    
                    # 计算统计信息
                    stats = {}
                    for col in columns:
                        if pd.api.types.is_numeric_dtype(standardized_df[col]):
                            stats[col] = {
                                'mean': float(standardized_df[col].mean()),
                                'std': float(standardized_df[col].std()),
                                'min': float(standardized_df[col].min()),
                                'max': float(standardized_df[col].max())
                            }
                    
                    # 转换为记录字典列表
                    records = convert_nan_to_none(standardized_df.to_dict('records'))
                    
                    # 构建结果
                    results[node_id] = {
                        "status": "success",
                        "message": f"数据标准化成功，处理了 {len(standardized_df)} 行数据",
                        "data": records,
                        "columns": columns,
                        "stats": stats,
                        "method": node_params.get('method', 'zscore'),
                        "feature_importance": columns
                    }
                    
                    # 将处理后的数据传递给后续节点
                    node_outputs[node_id] = {
                        "data": standardized_df,
                        "columns": columns,
                        "feature_importance": columns
                    }
                    
                    logger.info(f"数据标准化完成: {list(results[node_id].keys())}")
                    
                except Exception as e:
                    logger.exception(f"数据标准化失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"数据标准化失败: {str(e)}"
                    }
            elif node_type == 'one_hot_encoding':
                # OneHot编码节点
                if not upstream_data:
                    logger.error(f"OneHot编码节点 ID={node_id} 没有上游数据")
                    results[node_id] = {
                        "status": "error",
                        "message": "没有可用于编码的数据，请确保连接到了数据源节点"
                    }
                    continue
                
                try:
                    logger.info(f"OneHot编码节点 ID={node_id} 的上游数据: {list(upstream_data.keys())}")
                    
                    if 'data' not in upstream_data or upstream_data['data'] is None:
                        raise ValueError("上游节点没有提供数据")
                    
                    # 获取上游数据
                    if isinstance(upstream_data.get('data'), pd.DataFrame):
                        df = upstream_data.get('data')
                    elif isinstance(upstream_data.get('data'), list) and len(upstream_data.get('data', [])) > 0:
                        df = pd.DataFrame(upstream_data.get('data'))
                    else:
                        raise ValueError(f"上游数据格式不正确: {type(upstream_data.get('data'))}")
                    
                    logger.info(f"开始处理OneHot编码，数据形状: {df.shape}")
                    
                    # 执行OneHot编码
                    encoded_df = one_hot_encode(df, node_params)
                    
                    # 获取编码后的数据列
                    columns = encoded_df.columns.tolist()
                    
                    # 计算统计信息
                    stats = {}
                    for col in columns:
                        if pd.api.types.is_numeric_dtype(encoded_df[col]):
                            stats[col] = {
                                'mean': float(encoded_df[col].mean()),
                                'std': float(encoded_df[col].std()),
                                'min': float(encoded_df[col].min()),
                                'max': float(encoded_df[col].max())
                            }
                    
                    # 转换为记录字典列表
                    records = convert_nan_to_none(encoded_df.to_dict('records'))
                    
                    # 构建结果
                    results[node_id] = {
                        "status": "success",
                        "message": f"OneHot编码成功，处理了 {len(encoded_df)} 行数据",
                        "data": records,
                        "columns": columns,
                        "stats": stats,
                        "feature_importance": columns
                    }
                    
                    # 将编码后的数据传递给后续节点
                    node_outputs[node_id] = {
                        "data": encoded_df,
                        "columns": columns,
                        "feature_importance": columns
                    }
                    
                    logger.info(f"OneHot编码完成: {list(results[node_id].keys())}")
                    
                except Exception as e:
                    logger.exception(f"OneHot编码失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"OneHot编码失败: {str(e)}"
                    }
            elif node_type == 'feature_selection_filter':
                # 特征选择节点依赖上游数据
                if not upstream_data:
                    logger.error(f"特征选择节点 ID={node_id} 没有上游数据")
                    results[node_id] = {
                        "status": "error",
                        "message": "没有可用于特征选择的数据，请确保连接到了数据源节点"
                    }
                    continue
                
                try:
                    logger.info(f"特征选择节点 ID={node_id} 的上游数据: {list(upstream_data.keys())}")
                    
                    if 'data' not in upstream_data or upstream_data['data'] is None:
                        raise ValueError("上游节点没有提供数据")
                    
                    # 获取上游数据
                    if isinstance(upstream_data.get('data'), pd.DataFrame):
                        df = upstream_data.get('data')
                    elif isinstance(upstream_data.get('data'), list) and len(upstream_data.get('data', [])) > 0:
                        df = pd.DataFrame(upstream_data.get('data'))
                    else:
                        raise ValueError(f"上游数据格式不正确: {type(upstream_data.get('data'))}")
                    
                    logger.info(f"开始特征选择，数据形状: {df.shape}")
                    
                    # 执行特征选择
                    selected_df, selected_features, feature_scores = filter_features(df, node_params)
                    
                    # 转换为记录字典列表
                    records = convert_nan_to_none(selected_df.to_dict('records'))
                    
                    # 构建结果
                    results[node_id] = {
                        "status": "success",
                        "message": f"特征选择成功，保留了 {len(selected_features)} 个特征",
                        "data": records,
                        "columns": selected_features,
                        "selected_features": selected_features,
                        "original_features": df.columns.tolist(),
                        "feature_scores": feature_scores
                    }
                    
                    # 将处理后的数据传递给后续节点
                    node_outputs[node_id] = {
                        "data": selected_df,
                        "columns": selected_features,
                        "selected_features": selected_features,
                        "original_features": df.columns.tolist(),
                        "feature_scores": feature_scores
                    }
                    
                    logger.info(f"特征选择完成: {list(results[node_id].keys())}")
                    
                except Exception as e:
                    logger.exception(f"特征选择失败: {str(e)}")
                    results[node_id] = {
                        "status": "error",
                        "message": f"特征选择失败: {str(e)}"
                    }
            else:
                logger.warning(f"不支持的节点类型: {node_type}")
                results[node_id] = {
                    "status": "error", 
                    "message": f"不支持的节点类型: {node_type}"
                }
        except Exception as e:
            logger.exception(f"执行节点 ID={node_id} 时出错: {str(e)}")
            results[node_id] = {
                "status": "error", 
                "message": f"执行失败: {str(e)}"
            }
    
    return {"message": "节点执行完成", "results": results}

def topological_sort(nodes, edges_dict):
    """
    对节点进行拓扑排序，返回排序后的节点ID列表
    
    Args:
        nodes: 节点列表
        edges_dict: 边字典，键为目标节点ID，值为源节点ID列表
        
    Returns:
        排序后的节点ID列表，如果存在循环依赖则返回None
    """
    # 构建节点依赖关系图（反向边，为了方便得到节点的所有依赖）
    graph = {}
    for node in nodes:
        node_id = str(node.get('id'))
        graph[node_id] = []
    
    for target, sources in edges_dict.items():
        if target in graph:
            graph[target].extend(sources)
    
    # 计算每个节点的入度（有多少其他节点指向它）
    in_degree = {node_id: 0 for node_id in graph}
    for node_id, deps in graph.items():
        for dep in deps:
            if dep in in_degree:
                in_degree[dep] += 1
    
    # 找出入度为0的节点（没有依赖的节点）
    queue = [node_id for node_id, degree in in_degree.items() if degree == 0]
    result = []
    
    # 拓扑排序
    while queue:
        node_id = queue.pop(0)
        result.append(node_id)
        
        # 减少所有依赖当前节点的节点的入度
        for target, sources in edges_dict.items():
            if node_id in sources and target in in_degree:
                in_degree[target] -= 1
                if in_degree[target] == 0:
                    queue.append(target)
    
    # 如果结果列表长度小于节点数，说明存在循环依赖
    if len(result) < len(graph):
        return None
    
    # 由于我们的排序是基于依赖的，所以结果是按照从下游到上游的顺序
    # 需要反转结果，使其变为从上游到下游
    return result[::-1]

def evaluate_model(data: pd.DataFrame, params: dict) -> dict:
    """模型评估函数，评估机器学习模型的性能"""
    try:
        logger.info(f"模型评估参数: {json.dumps(params, default=str)}")
        
        # 检查上游数据是否包含metrics数据
        if not data or 'metrics' not in data:
            # 如果上游数据中没有metrics，检查是否有模型和测试数据可以用于评估
            if 'model' not in data or 'X' not in data or 'y' not in data:
                raise ValueError("上游数据中没有模型评估所需的数据")
            
            # 获取模型和测试数据
            model = data.get('model')
            X = data.get('X')
            y = data.get('y')
            
            # 分割数据
            from sklearn.model_selection import train_test_split
            test_size = float(params.get('test_size', 0.2))
            X_train, X_test, y_train, y_test = train_test_split(
                X, y, test_size=test_size, random_state=42
            )
            
            # 预测并评估
            y_pred = model.predict(X_test)
            
            # 计算评估指标
            from .ml_models import calculate_metrics
            metrics = calculate_metrics(y_test, y_pred, model, X_test)
            
            # 记录ROC曲线数据情况
            if 'roc_curve' in metrics:
                logger.info(f"模型评估生成ROC曲线数据: AUC={metrics['roc_curve']['auc']}")
            else:
                logger.info(f"模型评估未生成ROC曲线数据")
        else:
            # 直接使用上游数据中的metrics
            metrics = data.get('metrics')
            # 检查上游数据中是否有ROC曲线数据
            if 'roc_curve' in metrics:
                logger.info(f"从上游数据中获取到ROC曲线数据")
            else:
                logger.info(f"上游数据中没有ROC曲线数据")
        
        # 返回评估结果
        result = {
            'status': 'success',
            'message': '模型评估成功',
            'metrics': metrics
        }
        
        # 如果有特征重要性数据，添加到结果中
        if 'feature_importance' in data:
            result['feature_importance'] = data.get('feature_importance')
            logger.info(f"添加特征重要性数据到评估结果")
        
        return result
    except Exception as e:
        logger.exception(f"模型评估失败: {str(e)}")
        return {
            'status': 'error',
            'message': f'模型评估失败: {str(e)}',
            'data': None
        }

def one_hot_encode(data: pd.DataFrame, params: dict) -> pd.DataFrame:
    """对选择的分类特征执行OneHot编码"""
    try:
        logger.info(f"OneHot编码参数: {json.dumps(params, default=str)}")
        
        # 处理列选择
        columns_raw = params.get('columns', [])
        
        if isinstance(columns_raw, list):
            columns = columns_raw
        elif isinstance(columns_raw, str):
            columns = [col.strip() for col in columns_raw.split(',') if col.strip()]
        else:
            columns = []
        
        # 如果没有指定列，返回原始数据
        if not columns:
            logger.warning("未指定要编码的列")
            return data
        
        # 检查列是否存在于数据中
        valid_columns = [col for col in columns if col in data.columns]
        if not valid_columns:
            logger.warning(f"OneHot编码未找到有效列: {columns}")
            return data
        
        logger.info(f"开始OneHot编码，列: {valid_columns}")
        logger.info(f"原始数据形状: {data.shape}, 列: {list(data.columns)}")
        
        # 复制原始数据
        result = data.copy()
        original_columns = set(result.columns)
        
        # 处理编码前缀选项
        use_prefix = params.get('use_prefix', 'true') == 'true'
        logger.info(f"使用列名前缀: {use_prefix}")
        
        # 对每个分类列进行OneHot编码
        encoded_columns_count = 0
        
        for col in valid_columns:
            # 获取该列的独特值数量
            unique_values = data[col].nunique()
            logger.info(f"列 '{col}' 有 {unique_values} 个独特值")
            
            # 如果独特值数量过多，可以选择跳过或限制
            max_categories = int(params.get('max_categories', 20))
            if unique_values > max_categories:
                logger.warning(f"列 '{col}' 独特值数量 ({unique_values}) 超过限制 ({max_categories})，跳过编码")
                continue
            
            try:
                # 创建一个临时DataFrame只包含当前列
                temp_df = pd.DataFrame(data[col])
                
                # 执行OneHot编码
                prefix = col if use_prefix else None
                encoded = pd.get_dummies(temp_df, prefix=prefix, columns=[col], dtype=int)
                
                # 确保编码结果是整数类型
                for encoded_col in encoded.columns:
                    encoded[encoded_col] = encoded[encoded_col].astype(int)
                
                logger.info(f"列 '{col}' 编码后的列: {list(encoded.columns)}")
                logger.info(f"编码数据类型检查: {encoded.dtypes.to_dict()}")
                
                # 显示一部分编码后的数据确认是0和1
                sample_values = encoded.iloc[:5].to_dict() if len(encoded) > 0 else {}
                logger.info(f"编码样本数据（前5行）: {sample_values}")
                
                # 删除原始列，并连接编码后的列
                result = result.drop(col, axis=1)
                
                # 将编码列合并到结果中
                for enc_col in encoded.columns:
                    result[enc_col] = encoded[enc_col]
                    encoded_columns_count += 1
                
                logger.info(f"列 '{col}' OneHot编码成功，生成了 {len(encoded.columns)} 个新列")
            except Exception as e:
                logger.error(f"对列 '{col}' 进行OneHot编码时出错: {str(e)}")
        
        # 计算新增的列
        new_columns = set(result.columns) - original_columns
        logger.info(f"OneHot编码完成，新增列: {list(new_columns)}")
        logger.info(f"OneHot编码完成，原始数据有 {len(data.columns)} 列，编码后有 {len(result.columns)} 列")
        
        # 检查新增列的数据类型
        new_columns_dtypes = {col: result[col].dtype for col in new_columns}
        logger.info(f"新增列数据类型: {new_columns_dtypes}")
        
        # 确保所有编码后的列都是整数类型（0和1）
        for new_col in new_columns:
            if not pd.api.types.is_integer_dtype(result[new_col]):
                logger.warning(f"列 '{new_col}' 不是整数类型，正在转换...")
                result[new_col] = result[new_col].astype(int)
        
        # 更新结果中的列顺序，将新列放在前面以便检查
        if encoded_columns_count > 0:
            logger.info(f"编码后数据形状: {result.shape}, 前20列: {list(result.columns)[:20]}")
            
            # 记录编码后的部分数据，确认值是0和1
            encoded_sample = result[list(new_columns)].head(3).to_dict('records') if new_columns else {}
            logger.info(f"编码结果样本: {encoded_sample}")
        
        return result
    except Exception as e:
        logger.exception(f"OneHot编码失败: {str(e)}")
        return data

def filter_features(data: pd.DataFrame, params: dict) -> pd.DataFrame:
    """特征选择函数"""
    try:
        logger.info(f"特征选择参数: {json.dumps(params, default=str)}")
        
        # 获取特征选择方法
        method = params.get('feature_selection_method', 'variance')
        logger.info(f"特征选择方法: {method}")
        
        # 准备参数
        selection_params = {}
        
        # 根据方法设置参数
        if method == 'variance':
            threshold = params.get('variance_threshold', 0.0)
            selection_params['threshold'] = float(threshold)
            logger.info(f"方差过滤阈值: {threshold}")
        elif method == 'correlation':
            threshold = params.get('correlation_threshold', 0.9)
            selection_params['threshold'] = float(threshold)
            logger.info(f"特征间相关性过滤阈值: {threshold}")
        elif method == 'target_correlation':
            # 获取目标列
            target_column = params.get('target_column')
            if not target_column:
                logger.error("目标相关性过滤需要指定目标列")
                return data, data.columns.tolist(), {}
            
            # 获取选择方式
            selection_mode = params.get('selection_mode', 'threshold')
            
            if selection_mode == 'threshold':
                # 使用阈值
                threshold = params.get('threshold', 0.1)
                selection_params['threshold'] = float(threshold)
                logger.info(f"目标相关性过滤阈值: {threshold}")
            else:
                # 使用k
                k = params.get('k', 10)
                selection_params['k'] = int(k)
                logger.info(f"选择与目标相关性最高的特征数量: {k}")
                
            selection_params['target_column'] = target_column
            logger.info(f"目标相关性过滤目标列: {target_column}")
        elif method in ['chi2', 'f_test', 'mutual_info']:
            k = params.get('k_best', 10)
            selection_params['k'] = int(k)
            logger.info(f"选择特征数量: {k}")
        elif method == 'information_gain':
            # 获取目标列
            target_column = params.get('target_column')
            if not target_column:
                logger.error("信息增益特征选择需要指定目标列")
                return data, data.columns.tolist(), {}
                
            # 获取特征数量
            k = params.get('k_best', 10)
            selection_params['k'] = int(k)
            logger.info(f"信息增益选择特征数量: {k}")
            selection_params['target_column'] = target_column
            logger.info(f"信息增益特征选择目标列: {target_column}")
        elif method == 'relief':
            # 获取目标列
            target_column = params.get('target_column')
            if not target_column:
                logger.error("Relief算法需要指定目标列")
                return data, data.columns.tolist(), {}
                
            # 获取特征数量
            k = params.get('k_best', 10)
            selection_params['k'] = int(k)
            logger.info(f"Relief算法选择特征数量: {k}")
            
            # Relief特有参数：采样数量
            if params.get('sample_size'):
                selection_params['m'] = int(params.get('sample_size'))
                logger.info(f"Relief算法采样实例数量: {selection_params['m']}")
                
            selection_params['target_column'] = target_column
            logger.info(f"Relief算法目标列: {target_column}")
        
        # 获取目标列
        target_column = params.get('target_column')
        
        # 如果是监督学习的特征选择方法，但没有提供目标列，则报错
        if method in ['chi2', 'f_test', 'mutual_info', 'information_gain', 'relief'] and not target_column:
            logger.error(f"方法 {method} 需要提供目标列，但未指定")
            return data, data.columns.tolist(), {}
        
        # 执行特征选择
        selected_df, selected_features, feature_scores = feature_selection_core(
            data,
            method=method,
            params=selection_params,
            target_column=target_column
        )
        
        logger.info(f"特征选择完成: 从 {data.shape[1]} 个特征选择了 {len(selected_features)} 个特征")
        
        # 返回结果
        return selected_df, selected_features, feature_scores
    except Exception as e:
        logger.exception(f"特征选择失败: {str(e)}")
        return data, data.columns.tolist(), {}